package cn.itcast.flink.transformation;

import com.alibaba.fastjson.JSON;
import lombok.Data;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author lilulu
 */
//需求：将读取文本文件数据，每行JSON格式数据，转换为ClickLog对象，使用 map 函数完成。
public class TransformationBasicDemo {

    @Data
    public static class ClickLog {
        //频道ID
        private long channelId;
        //产品的类别ID
        private long categoryId;
        //产品ID
        private long produceId;
        //用户的ID
        private long userId;
        //国家
        private String country;
        //省份
        private String province;
        //城市
        private String city;
        //网络方式
        private String network;
        //来源方式
        private String source;
        //浏览器类型
        private String browserType;
        //进入网站时间
        private Long entryTime;
        //离开网站时间
        private Long leaveTime;
    }

    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<String> source = env.readTextFile("datas/click.log");
        // 3. 数据转换-transformation
        // TODO: 函数一【map函数】，将JSON转换为JavaBean对象
        SingleOutputStreamOperator<ClickLog> operator = source.map(
                new MapFunction<String, ClickLog>() {
                    public ClickLog map(String line) throws Exception {
                        return JSON.parseObject(line, ClickLog.class);
                    }
                }
        );
        // 4. 数据终端-sink
//        operator.printToErr();

        // TODO: 2022/7/19 [flatMap 算子】 -> 每条数据中Long类型值的时间字段转换为不同日期时间格式字符串
        /*
            Long类型日期时间：	1577890860000
                            |
                            |进行格式
                            |
            String类型日期格式
                    yyyy-MM-dd-HH
                    yyyy-MM-dd
                    yyyy-MM
            todo 使用工具类：
                DateFormatUtils, 在commons.langs 包下面
         */
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = operator.flatMap(
                new FlatMapFunction<ClickLog, String>() {
                    public void flatMap(ClickLog clickLog, Collector<String> collector) throws Exception {
                        Long entryTime = clickLog.getEntryTime();
                        String hourTime = DateFormatUtils.format(entryTime, "yyyy-MM-dd-HH");
                        collector.collect(hourTime);
                        String dayTime = DateFormatUtils.format(entryTime, "yyyy-MM-dd");
                        collector.collect(dayTime);
                        String monthTime = DateFormatUtils.format(entryTime, "yyyy-MM");
                        collector.collect(monthTime);
                    }
                }
        );
//        stringSingleOutputStreamOperator.printToErr();


        // TODO: 2022/7/19 [filter 算子]， 过滤获取使用谷歌浏览器访问日志数据
       /* SingleOutputStreamOperator<ClickLog> filterDataStream = operator.filter(
                new FilterFunction<ClickLog>() {
                    public boolean filter(ClickLog clickLog) throws Exception {
                        return "谷歌浏览器".equals(clickLog.browserType);
                    }
                });
        filterDataStream.printToErr();*/
        SingleOutputStreamOperator<ClickLog> filter = operator.filter(clickLog -> "谷歌浏览器".equals(clickLog.browserType));
        filter.printToErr();
        // 5. 触发执行-execute
        env.execute("TransformationBasicDemo");
    }
}